package com.atguigu.dga.governance.assessor.calc;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.common.utils.SqlParser;
import com.atguigu.dga.ds.bean.TDsTaskDefinition;
import com.atguigu.dga.governance.assessor.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.google.common.collect.Sets;
import lombok.Getter;
import lombok.Setter;
import org.antlr.runtime.tree.Tree;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.ql.lib.Dispatcher;
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.parse.ASTNode;
import org.apache.hadoop.hive.ql.parse.HintParser;
import org.apache.hadoop.hive.ql.parse.HiveParser;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.text.ParseException;
import java.util.*;
import java.util.stream.Collectors;

@Component("IS_SIMPLE_PROCESS")
public class IsSimpleProcessAssessor extends Assessor {

    //1   提取、解析sql
    //2   制作一个节点处理器
    //     2.1   收集是否有复杂的处理特征
    //     2.2   收集where 条件涉及的字段名
    //     2.3   收集来源表
    //3  遍历结束后  从节点处理器提取收集的信息
    //4  结合元数据信息 对 sql进行定性
    //5  根据结论评分
    @Override
    public void checkProblem(AssessParam assessParam, GovernanceAssessDetail governanceAssessDetail) throws ParseException {
        TableMetaInfo tableMetaInfo = assessParam.getTableMetaInfo();

        TDsTaskDefinition tDsTaskDefinition = assessParam.getTDsTaskDefinition();
        if(tDsTaskDefinition==null){
            return;
        }
        String sql = tDsTaskDefinition.getSql();
        SimpleProcessDispatcher simpleProcessDispatcher = new SimpleProcessDispatcher();
        try {
            SqlParser.parseSQL(sql, simpleProcessDispatcher);
        } catch (Exception e) {
            throw new RuntimeException("sql 解析异常");
        }
        Set<String> complicationProcessTokens = simpleProcessDispatcher.getComplicationProcessTokens();
        Set<String> sourceTableNames = simpleProcessDispatcher.getSourceTableNames();
        Set<String> whereConditionFieldNames = simpleProcessDispatcher.getWhereConditionFieldNames();
        System.out.println("whereConditionFieldNames = " + whereConditionFieldNames);
        // 如果有任何复杂处理都不需要继续判断 都直接放过
        if(complicationProcessTokens.size()==0){
            //依据表名提取对应的元数据信息
            List<TableMetaInfo> tableMetaInfoList = assessParam.getTableMetaInfoList();
            Set<String> nonPartitionColNameSet = new HashSet<>();//存放非分区字段
            Map<String, TableMetaInfo> sourceTableMetaInfoMap = tableMetaInfoList.stream().collect(Collectors.toMap(sourceTableMetaInfo -> sourceTableMetaInfo.getSchemaName() + "." + sourceTableMetaInfo.getTableName(), sourceTableMetaInfo -> sourceTableMetaInfo));
            for (String sourceTableName : sourceTableNames) {
                TableMetaInfo sourceTableMetaInfo = sourceTableMetaInfoMap.get(sourceTableName);
                if(sourceTableMetaInfo!=null){
                    String partitionColNameJson = sourceTableMetaInfo.getPartitionColNameJson();
                    //对比是否有非分区字段
                    List<JSONObject> partitionColNameJsonList = JSON.parseArray(partitionColNameJson, JSONObject.class);
                    Set<String> partitionColNameSet = partitionColNameJsonList.stream().map(jsonObject -> jsonObject.getString("name")).collect(Collectors.toSet());
                    for (String whereConditionFieldName : whereConditionFieldNames) {
                        if (!partitionColNameSet.contains(whereConditionFieldName)) { //如果不属于分区字段 则加入非分区字段集合
                            nonPartitionColNameSet.add(whereConditionFieldName);
                        }
                    }
                }
            }
            if(nonPartitionColNameSet.size()==0){ //既不是复杂操作  where中又全都是分区字段 则定性为简单加工
                governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
                governanceAssessDetail.setAssessProblem("不存在复杂操作，where 字段 不存在非分区字段");
            }else{
                governanceAssessDetail.setAssessComment("sql 中包含非分区字段："+ StringUtils.join(",",nonPartitionColNameSet) );
            }

        }else{
            governanceAssessDetail.setAssessComment(" sql 中包含复杂操作："+ StringUtils.join(",",complicationProcessTokens) );
        }


    }



    //    //     2.1   收集是否有复杂的处理特征
    //    //     2.2   收集where 条件涉及的字段名
    //    //     2.3   收集来源表
    class  SimpleProcessDispatcher implements Dispatcher {

        //收集是否有复杂的处理特征
        @Getter
        Set<String> complicationProcessTokens = new HashSet<>();
        //收集where 条件涉及的字段名
        @Getter
        Set<String> whereConditionFieldNames = new HashSet<>();
        //收集来源表
        @Getter
        Set<String> sourceTableNames = new HashSet<>();

        @Setter
        String defaultSchemaName;

        Set<Integer>   complicateTokenTypeSet = Sets.newHashSet(HiveParser.TOK_JOIN,  //join 包含通过where 连接的情况
                HiveParser.TOK_GROUPBY,       //  group by
                HiveParser.TOK_LEFTOUTERJOIN,       //  left join
                HiveParser.TOK_RIGHTOUTERJOIN,     //   right join
                HiveParser.TOK_FULLOUTERJOIN,     // full join
                HiveParser.TOK_FUNCTION,     //count(1) sum(a)
                HiveParser.TOK_FUNCTIONDI,  //count(distinct xx)
                HiveParser.TOK_FUNCTIONSTAR, // count(*)
                HiveParser.TOK_SELECTDI,  // distinct
                HiveParser.TOK_UNIONALL,   // union
                HiveParser.TOK_LATERAL_VIEW ,
                HiveParser.TOK_LATERAL_VIEW_OUTER
        );


        @Override
        public Object dispatch(Node nd, Stack<Node> stack, Object... nodeOutputs) throws SemanticException {
            ASTNode astNode = (ASTNode) nd;
            //1 采集 哪些token属于复杂操作
            if(complicateTokenTypeSet.contains(astNode.getType()) ){
                complicationProcessTokens.add(astNode.getText());
            }
            //2 收集来源表  // 要带库名  // 如果没有库名 则给一个默认库名 给考评表当前库名
            if(astNode.getType()==HiveParser.TOK_TABREF){
                ASTNode tokTablenameNode = (ASTNode) astNode.getChild(0);
                if(tokTablenameNode.getChildCount()==2){  //带库名的情况
                    String schemaName = tokTablenameNode.getChild(0).getText();
                    String tableName = tokTablenameNode.getChild(1).getText();
                    sourceTableNames.add(schemaName+"."+tableName);
                }else{  //不带库名
                    String schemaName = defaultSchemaName;
                    String tableName = tokTablenameNode.getChild(0).getText();
                    sourceTableNames.add(schemaName+"."+tableName);
                }

            }
            //3  收集where 条件涉及的字段名   //判断逻辑：是字段TOK_TABLE_OR_COL 而且是where子句中 不是 select子句中的
            if(astNode.getType()== HiveParser.TOK_TABLE_OR_COL&&astNode.hasAncestor(HiveParser.TOK_WHERE)&&!astNode.hasAncestor(HiveParser.TOK_SELECT)){
                // 要分辨 是否字段带表名，如果 TOK_TABLE_OR_COL的父节点为.  则从兄弟节点中取字段名 ，否则从子节点取字段名
               if(astNode.getParent().getType()==HiveParser.DOT){
                   ASTNode fieldNameNode =(ASTNode) astNode.getParent().getChild(1);
                   whereConditionFieldNames.add(fieldNameNode.getText())  ;
               }else{
                   String fieldName = astNode.getChild(0).getText();
                   whereConditionFieldNames.add(fieldName);
               }

            }

            return null;
        }
    }

}
